package kafka;

import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;

/**
 * Create by zhh on 2022/9/19.
 *
 */
public class flinkUtils {

    public static StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    public static  DataStream<String> createKafkaStreamV1(String[] args) throws IOException, IllegalAccessException, InstantiationException {

        ParameterTool tool = ParameterTool.fromPropertiesFile(args[0]);
        String groupId = tool.get("group.id","test");//不填则选择后面的默认值
        String servers = tool.getRequired("bootstrap.servers");//必填
        List<String> topics = Arrays.asList(tool.getRequired("kafka.input.topics").split(","));//可能有多个
        String autoCommit = tool.get("enable.auto.commit", "false");
        String offsetReset = tool.get("auto.offset.reset", "earliest");
        //kafka相关参数
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", servers);
        properties.setProperty("group.id", groupId);
        properties.setProperty("enable.auto.commit",autoCommit);//是否自动提交，默认为true，设置为false，存在状态中
        properties.setProperty("auto.offset.reset",offsetReset);//没有offset就从最早开始消费，如果有则从上次挂掉的地方开始消费（状态中）


        //checkpoint
        int checkpointInterval = tool.getInt("checkpoint.interval", 5000);
        String checkpointPath = tool.get("checkpoint.path", "hdfs://node01:9000/Flink/checkpoint");


        env.enableCheckpointing(checkpointInterval);
        env.setStateBackend(new FsStateBackend(checkpointPath));
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);//保留checkpoint文件
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(2, Time.seconds(5)));//重启策略 重启两次，间隔5秒

        //Kafka的数据
        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(topics, new SimpleStringSchema(), properties);

        return env.addSource(kafkaConsumer);
    }
    public static <T> DataStream<T> createKafkaStreamV2(String[] args, Class<? extends DeserializationSchema<T>> deser) throws IOException, IllegalAccessException, InstantiationException {

        ParameterTool tool = ParameterTool.fromPropertiesFile(args[0]);
        String groupId = tool.get("group.id","test");//不填则选择后面的默认值
        String servers = tool.getRequired("bootstrap.servers");//必填
        List<String> topics = Arrays.asList(tool.getRequired("kafka.input.topics").split(","));//可能有多个
        String autoCommit = tool.get("enable.auto.commit", "false");
        String offsetReset = tool.get("auto.offset.reset", "earliest");
        //kafka相关参数
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", servers);
        properties.setProperty("group.id", groupId);
        properties.setProperty("enable.auto.commit",autoCommit);//是否自动提交，默认为true，设置为false，存在状态中
        properties.setProperty("auto.offset.reset",offsetReset);//没有offset就从最早开始消费，如果有则从上次挂掉的地方开始消费（状态中）


        //checkpoint
        int checkpointInterval = tool.getInt("checkpoint.interval", 5000);
        String checkpointPath = tool.get("checkpoint.path", "hdfs://node01:9000/Flink/checkpoint");


        env.enableCheckpointing(checkpointInterval);
        env.setStateBackend(new FsStateBackend(checkpointPath));
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);//保留checkpoint文件
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(2, Time.seconds(5)));//重启策略 重启两次，间隔5秒

        //Kafka的数据
        FlinkKafkaConsumer<T> kafkaConsumer = new FlinkKafkaConsumer<T>(topics, deser.newInstance(), properties);

        return env.addSource(kafkaConsumer);
    }

}
